package com.imooc.scala

import java.io.File

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object DataFrameRDDApp {
  def main(args: Array[String]): Unit = {
    //    System.setProperty("hadoop.home.dir", "C:\\workspace\\scala\\winutils\\hadoop-2.6.0")
    val spark = SparkSession.builder().appName("Data").master("local[*]").getOrCreate()
    val rdd = spark.sparkContext.textFile("infos.txt")
    rdd.foreach(println)
    //    方法1 知道结构,先定义一个case class
    //    import spark.implicits._
    //    val infoDF = rdd.map(_.split(",")).map(line => Info(line(0).toInt, line(1), line(2).toInt)).toDF()
    //
    //    infoDF.show()
    //    infoDF.filter(infoDF.col("age") > 30).show()
    //    infoDF.createOrReplaceTempView("infos")
    //    spark.sql("select * from infos where age>30").show()

    //    方法2
    val infoRDD = rdd.map(_.split(",")).map(line => Row(line(0).toInt, line(1), line(2).toInt))
    val strucType = StructType(Array(StructField("id", IntegerType, true), StructField("name", StringType, true), StructField("age", IntegerType, true)))

    val infoDF2 = spark.createDataFrame(infoRDD, strucType)
    infoDF2.show()

    infoDF2.select("name").show //write.json("file:///c:/workspace/scala/scala-train")

    val dir = "c:/workspace/scala/scala-train/json"
    deleteDir(new File(dir))
    infoDF2.write.json(dir)

    spark.stop()
  }

  case class Info(id: Int, name: String, age: Int)

  def deleteDir(dir: File): Unit = {
    val files = dir.listFiles()
    files.foreach(f => {
      if (f.isDirectory) {
        deleteDir(f)
      } else {
        f.delete()
        println("delete file " + f.getAbsolutePath)
      }
    })
    dir.delete()
    println("delete dir " + dir.getAbsolutePath)
  }
}
